package org.suyafei.func;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.suyafei.util.TimeUtil;

/**
 * @author 宿某人
 * @Date: 2022/03/29/11:32
 * @Description:
 */
@Slf4j
public class VisitorMapFunction extends RichMapFunction<JSONObject, String> {

    /**
     * 记录用户端udid
     */
    private ValueState<String> newDayVisitorState;

    @Override
    public void open(Configuration parameters) throws Exception {
         newDayVisitorState = getRuntimeContext().getState(new ValueStateDescriptor<String>("newDayVisitorState", String.class));
    }

    @Override
    public String map(JSONObject jsonObject) throws Exception {

        String beforeDateState = newDayVisitorState.value();

        Long ts = jsonObject.getLong("ts");

        String currentDateStr = TimeUtil.format(ts);
        //判断日期是否为空，进行新老访客的识别
        if (StringUtils.isNotEmpty(beforeDateState)) {
            if (beforeDateState.equalsIgnoreCase(currentDateStr)) {
                //一样则是老访客
                jsonObject.put("is_new", 0);
                log.info("老访客:{}",currentDateStr);
            }else {
                jsonObject.put("is_new", 1);
                newDayVisitorState.update(currentDateStr);
                log.info("新访客：{}",currentDateStr);
            }
        }else {
            //如果状态为空，则是新用户，标记1，老访客标记0
            jsonObject.put("is_new", 1);
            newDayVisitorState.update(currentDateStr);
            log.info("新访客:{}", currentDateStr);
        }
        return jsonObject.toJSONString();
    }
}
